package me.seawenc.datastash.outputs.jdbc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import me.seawenc.datastash.outputs.AbstractConsumer;
import me.seawenc.datastash.outputs.jdbc.helper.SqlActuator;

public class JdbcConsumer extends AbstractConsumer {
    private final String url;
    private final String username;
    private final String password;
    private final String driver;
    private final String insertSql;
    
    // 只支持csv和json格式
    private final  String dataFormat;

    final private List<String> jsonPaths;

    final SqlActuator sqlActuator;

    public JdbcConsumer(Properties props) {
        super(props);
        url = props.getProperty("output.jdbc.url");
        username = props.getProperty("output.jdbc.username");
        password = props.getProperty("output.jdbc.password");
        driver = props.getProperty("output.jdbc.driver");
        insertSql=props.getProperty("output.jdbc.sql").replace(";","");
        if(!insertSql.startsWith("insert")){
            throw new RuntimeException("output.jdbc.insert.sql只支持insert操作");
        }
        dataFormat=props.getProperty("output.jdbc.dataFormat").toLowerCase();
        if(!dataFormat.equals("csv") && !dataFormat.equals("json")){
            throw new RuntimeException("output.jdbc.dataFormat只支持csv和json");
        }
        jsonPaths=findJsonPaths(insertSql);
        sqlActuator=new SqlActuator(url,username,password,driver);
    }

    @Override
    public void consumption(String msg) {
        if(msg==null|| "".equals(msg.trim())){
            return;
        }
        JSONObject json=null;
        if("csv".equals(dataFormat)){
            json=new JSONObject();
            String[] fields = msg.split(",");
            for(int i=0;i<fields.length;i++){
                json.put("f"+i,fields[i]);
            }
        }else{
            if("json".equals(dataFormat)){
                json= JSON.parseObject(msg);
            }
        }

        String sql = replaceMsg(json);
        try {
            sqlActuator.execInsert(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private String replaceMsg(JSONObject jsonMsg) {
        String currentInsertSql=insertSql.toString();
        for(String jsonPath:jsonPaths){
            String string = jsonMsg.getString(jsonPath);
            String format = String.format("~:%s:~", jsonPath);
            if(null==string ||"".equals(string)){
                throw new RuntimeException("配置的字段不存在:"+format);
            }
            currentInsertSql=currentInsertSql.replace(format, string);
        }
        return currentInsertSql;
    }

    private List<String> findJsonPaths(Object msgTemp) {
        String tmp=msgTemp.toString();
        Pattern pattern = Pattern.compile("~:([^(~:)&^(:~)]+):~");
        Matcher matcher = pattern.matcher(tmp);
        List<String> paths=new ArrayList<>();
        while (matcher.find()) {
            String variable = matcher.group(1);
            paths.add(variable);
        }
        return paths;
    }
}
